package xin.marcher.wind.migrate.component.canal;

import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import xin.marcher.wind.migrate.domain.entity.ScrollDomainDO;
import xin.marcher.wind.migrate.service.MigrateConfigService;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * canal mq 消息消费任务
 */
@Component
public class CanalConsumeTask implements ApplicationRunner {

    @Resource
    private RocketMQProperties rocketMQProperties;

    @Resource
    private MigrateConfigService migrateConfigService;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        String mqNameServer = rocketMQProperties.getNameServer();

        // 当前配置的全部迁移系统
        List<ScrollDomainDO> scrollDomainDOList = migrateConfigService.queryScrollDomainList();

        // 线程数量与源库表一致
        ExecutorService executors = Executors.newFixedThreadPool(scrollDomainDOList.size());
        for (ScrollDomainDO scrollDomainDO : scrollDomainDOList) {
            if (scrollDomainDO.getDataSourceType().equals(1)) {
                /*
                * 从 rocketmq 里拉取 binlog 的时候，避免让 consumer 自动去提交 offset
                * 精准的去控制 offset 提交，要确保每一条 binlog 都已经被应用到目标数据源里去了, 此时对这条 offset 才能去做一个提交,
                */
                // 执行拉取任务
                executors.execute(new CanalPullRunner(scrollDomainDO.getDomainTopic(), mqNameServer));
                // 执行提交任务
                executors.execute(new CanalPullCommitRunner(scrollDomainDO.getDomainTopic(), mqNameServer));
            }
        }
    }
}
